package tk.xboot.flink.cep.le;

import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.Arrays;
import java.util.List;
import java.util.Map;

/**
 *
 * To use flink cep, please add the fink-cep-scala_2.11 to your pom.mxl.
 *
 *
 *
 * <pre>
 *     用户登录日志格式如下表:
 *     时间                         用户编号            IP地扯             登录类型
 *      2018-11-19T12:12:12           1              192.168.0.1         fail
 *      2018-11-19T12:12:12           1              192.168.0.1         fail
 *      2018-11-19T12:12:12           1              192.168.0.1         fail
 *      2018-11-19T12:12:12           2              192.168.10,10      success
 *
 * </pre>
 *
 * We need define a bean to describe the login event, called LoginEvent.
 *
 *
 *
 *
 */
public class LoginFailedCep {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<LoginEvent> loginEventDataStream = env.fromCollection(Arrays.asList(
                new LoginEvent("1","192.168.0.1","fail"),
                new LoginEvent("1","192.168.0.2","fail"),
                new LoginEvent("1","192.168.0.3","fail"),
                new LoginEvent("2","192.168.10.10","success")
        ));

        Pattern<LoginEvent, LoginEvent> loginFailPattern = Pattern.<LoginEvent>
                begin("begin")
                .where(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent, Context<LoginEvent> context) throws Exception {
                        return loginEvent.getType().equals("fail");
                    }
                }).next("next")
                .where(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent, Context<LoginEvent> context) throws Exception {
                        return loginEvent.getType().equals("fail");
                    }
                }).within(Time.milliseconds(3));


        PatternStream<LoginEvent> patternStream = CEP.pattern(
                loginEventDataStream.keyBy(LoginEvent::getUserId),loginFailPattern
        );
        DataStream<LoginWarning> loginFailDataStream = patternStream
                .select((Map<String,List<LoginEvent>> pattern)->{
                    List<LoginEvent> first = pattern.get("begin");
                    List<LoginEvent> second = pattern.get("next");
                    LoginEvent le = second.get(0);
                    return new LoginWarning(le.getUserId(),le.getType(),le.getIp());
                });

        loginEventDataStream.print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
